package cn.hesion.friend.processor;
/**
 * ClassName: RelationDeleteProcessor <br/>
 * Description: <br/>
 * date: 2021/3/13 9:52<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

/**
 * 删除好友协处理器
 *
 * @author: hesion
 * @create: 2021-03-13 09:52
 **/
public class RelationDeleteProcessor extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //获取到表的操作对象
        HTableInterface relation = e.getEnvironment().getTable(TableName.valueOf("relation"));
        //获取RowKey
        byte[] row = delete.getRow();
        //获取到了谁删除了谁
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        Set<Map.Entry<byte[], List<Cell>>> entries = familyCellMap.entrySet();
        for (Map.Entry<byte[], List<Cell>> entry : entries) {
            //列族信息
            System.out.println(Bytes.toString(entry.getKey()));
            //该列族中的所有单元格
            List<Cell> cells = entry.getValue();
            for (Cell cell : cells) {
                //获取具体单元格的rowkey
                byte[] rowKey = CellUtil.cloneRow(cell);
                //获取具体的列
                byte[] column = CellUtil.cloneQualifier(cell);

                //删除之前判断下数据是否存在
                boolean flag = relation.exists(new Get(column).addColumn(Bytes.toBytes("friends"), rowKey));

                if (flag) {
                    Delete friend = new Delete(column).addColumn(Bytes.toBytes("friends"), rowKey);
                    relation.delete(friend);
                }
            }
        }
    }
}
